package com.etc

import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
/**
  * @Title:
  * @ProjectName
  * @Description:
  * @author kalista
  */


object KafkaProducer {
  //kafka节点
  def BROKER_LIST = "hdc-data4:9092,hdc-data5:9092,hdc-data6:9092"

  def TOPIC = "topicA"

  def isAsync = false

  def main(args: Array[String]): Unit = {

    val props = new Properties()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST)
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer")
    val producer = new KafkaProducer[Int, String](props)
    try {
      //模拟三个电表产生电量数据 active_quan:电量，create_time:电量产生时间，meter_id:电表id
      while (true) {
        val cur_time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date)
        val arr = Array("000a3804-0315-4de4-a846-74ac37812d08", "010892ae-9d99-42cf-aab7-6d074351b15a", "01e5a01c-661d-4b65-82ee-1309497b79e7")
        for (meter_id <- arr) {
          val msg = "{\"active_quan\":" + ((new util.Random).nextInt(100) + 1) + ",\"create_time\":\"" + cur_time + "\",\"meter_id\":\"" + meter_id + "\"}"
          val startTime = System.currentTimeMillis()
          if (isAsync) {
            // Send asynchronously
            producer.send(new ProducerRecord(TOPIC, msg), new DemoCallback(startTime, msg))
          } else {
            // Send synchronously
            producer.send(new ProducerRecord(TOPIC,msg))
            System.out.println("Sent message: (" + msg + ")")
          }
        }
        Thread.sleep(30000)
      }
    } catch {
      case ex: Exception => {
        println(ex)
      }
    } finally {
      producer.close
    }
  }
}
